package com.ibm.cps.api;


import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import javax.inject.Named;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo;

import com.ibm.cps.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.jarxs.DatasourceResource;
import com.ibm.cps.kafka.KafkaProducer;
import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

@Named
public class DatasourceBean implements DatasourceResource{


	private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();
	private Logger logger;
	private KafkaProducer producer;
	private IPersistency persistency;
	private DataSourceMonitor monitor;
	private AuthorizationManager authManager;

	public DatasourceBean() throws CPSException {
		monitor = DataSourceMonitor.getInstance();
		logger = LoggerFactory.getLogger(this.getClass());
		producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
		persistency = MongoPersistency.getInstace();
		authManager = AuthorizationManager.getInstance();
		logger.info("Connect to kafka: " + LocalConfig.KAFKA_BROKERS);
		logger.info("Connect to mongo: " + LocalConfig.MONGO_CONNECT_STRING + ", and database is "
				+ LocalConfig.PERSISTENCY_DATABASE);
	}

	@Override
	public Response GetDatasources(String tenantId, UriInfo info,String auth) {

//		GET /DataSources/<tenantId>?<field_name>=<query_value>
//		特定字段检索

		if (!authManager.isAllow(tenantId, auth)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}

		ObjectNode filter = JSONUtility.newObjectNode(MessageFields.TENANTID,tenantId);

		MultivaluedMap<String, String> map = info.getQueryParameters();
		if (map.size() > 0) {
			for(String key : map.keySet()){
				filter.put(key, map.getFirst(key));
			}
		}

		String source = info.getQueryParameters().getFirst(MessageFields.SOURCE);
		if (source == null) {
			source = "rest";
		}

		String finalstr = "";
		try {
			PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantId);
			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantId), publishFactory, source);
			Collection<String> metadatas = persistency.getTenantDataSource(filter);
			Collection<String> result = adapter.getJsons(metadatas, metadatas);
			finalstr = ServletUtil.constructCollectionJson(result);
		} catch (CPSException e) {
			String ret = ServletUtil.constructReturnMsg(400, e.getErrorMessage());
			return Response.ok().entity(ret).type(MediaType.APPLICATION_JSON).build();
		}

		return Response.ok(finalstr).type(MediaType.APPLICATION_JSON).build();
	}

	@Override
	public Response GetDatasource(String tenantId, String datasource_name,String auth) {

		if (!authManager.isAllow(tenantId, auth)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}

		String source = "rest";
		String finalstr = "";
		try {
			PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantId);
			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantId), publishFactory, source);
			Collection<String> metadatas = persistency.getTenantDataSource(tenantId, datasource_name);
			Collection<String> result = adapter.getJsons(metadatas, metadatas);
			finalstr = JSONUtility.toJsonString(result);
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

		return Response.ok(finalstr).type(MediaType.APPLICATION_JSON).build();

	}

	@Override
	public Response PutDatasource(String tenantId, String ds,String auth) {
		String ret = "";
		if (!authManager.isAllow(tenantId, auth)) {
			ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		ret = ServletUtil.constructReturnMsg(ErrorCode.NO_IMPLEMENTATION_CODE, "To be implement");
		return Response.ok().entity(ret).type(MediaType.APPLICATION_JSON).build();
	}

	@Override
	public Response PostDatasource(String tenantId,String json,String auth) {
		String ret = "";
		if (!authManager.isAllow(tenantId, auth)) {
			ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}

		if (json == null || json.length() < 1) {
			ret = ServletUtil.constructReturnMsg(ErrorCode.INVALID_INPUT_ERROR, "Invalid input, please check.");
			return Response.ok().entity(ret).type(MediaType.APPLICATION_JSON).build();
		}
		ArrayList<AbstractDataSourceMetadata> sendObjects = new ArrayList<AbstractDataSourceMetadata>();
		try {
			ArrayNode array = JSONUtility.getArrayFromString(json);
			ArrayNode retArray = JSONUtility.newArrayNode();

			for (int i = 0; i < array.size(); i++) {
				ObjectNode objectNode = (ObjectNode) array.get(i);
				AbstractDataSourceMetadata dataSource = monitor.validateDataSource(tenantId, objectNode);
				ObjectNode retNode = JSONUtility.newObjectNode();
				retNode.put("id", dataSource.getName());
				retArray.add(retNode);
				sendObjects.add(dataSource);
			}

			for (int i = 0; i < sendObjects.size(); i++) {
				persistency.putDataSource(sendObjects.get(i).getInput());
				String type = sendObjects.get(i).getType();
				if (DataSourceMetadataFactory.isStaticDataSource(type)) {
					// for static datasource
					// do nothing.
				}else {
					// for graphite and mqtt
					producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantId),
							sendObjects.get(i).getInput().getBytes());
				}
			}

			String result = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retArray);
			return Response.ok(result).type(MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			try {
				for (int i = 0; i < sendObjects.size(); i++) {
					AbstractDataSourceMetadata tmpObject = sendObjects.get(i);
					persistency.deleteDataSource(tenantId, tmpObject.getName());
					monitor.deleteDataSource(tenantId, tmpObject);
				}
				throw e;
			} catch (CPSException e1) {
				e1.printStackTrace();
			}
			ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

	}

	@Override
	public Response DeleteDatasources(String tenantId,String auth) {
		if (!authManager.isAllow(tenantId, auth)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}

		try {
			Collection<String> metadatas = persistency.getTenantDataSource(tenantId,null);
			if (metadatas == null) {
				String str =  ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted");
				return Response.ok(str).type(MediaType.APPLICATION_JSON).build();
			}

			for (String dataSource : metadatas) {
				AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
				String name = metadata.getName();
				ObjectNode node = JSONUtility.newObjectNode();
				node.set(MessageFields.PARENTPROCESSORS, JSONUtility.newArrayNode().add(name));

				Collection<String> inputs = persistency.getProcessors(node);
				if (inputs != null && inputs.size() >= 1) {
					throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Can't delete this data source "
							+ metadata.getName() + " because it's referenced by other processor metadatas.");
				}
			}
			List<String> deleted = new ArrayList<String>();

			try {
				for (String dataSource : metadatas) {
					AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
					monitor.deleteDataSource(tenantId, metadata);
					Collection<String> result = persistency.deleteDataSource(tenantId, metadata.getName());
					producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceDeleteTopic(tenantId),
							metadata.getInput().getBytes());

					if (result != null) {
						for (String ds : result) {
							deleted.add(ds);
						}
					}
				}
			} catch (Throwable e) {

				// transactionality
				for (String dataSource : deleted) {
					monitor.validateDataSource(tenantId, JSONUtility.fromObjectString(dataSource));
					persistency.putDataSource(dataSource);
					producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantId), dataSource.getBytes());
				}
				throw e;
			}
			String str = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, ServletUtil.constructCollectionJson(deleted));
			return Response.ok(str).type(MediaType.APPLICATION_JSON).build();

		} catch (Exception e) {
			e.printStackTrace();
		} catch (Throwable e) {
			e.printStackTrace();
		}
		String ret = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "");
		return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();

	}

	@Override
	public Response DeleteDatasource(String tenantId, String ds_name,String auth) {
		if (!authManager.isAllow(tenantId, auth)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}

		try {
			Collection<String> metadatas = persistency.getTenantDataSource(tenantId, ds_name);
			if (metadatas == null) {
				String str =  ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted");
				return Response.ok(str).type(MediaType.APPLICATION_JSON).build();
			}

			for (String dataSource : metadatas) {
				AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
				String name = metadata.getName();
				ObjectNode node = JSONUtility.newObjectNode();
				node.set(MessageFields.PARENTPROCESSORS, JSONUtility.newArrayNode().add(name));

				Collection<String> inputs = persistency.getProcessors(node);
				if (inputs != null && inputs.size() >= 1) {
					throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Can't delete this data source "
							+ metadata.getName() + " because it's referenced by other processor metadatas.");
				}
			}
			List<String> deleted = new ArrayList<String>();

			try {
				for (String dataSource : metadatas) {
					AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
					monitor.deleteDataSource(tenantId, metadata);
					Collection<String> result = persistency.deleteDataSource(tenantId, metadata.getName());
					producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceDeleteTopic(tenantId),
							metadata.getInput().getBytes());

					if (result != null) {
						for (String ds : result) {
							deleted.add(ds);
						}
					}
				}
			} catch (Throwable e) {

				// transactionality
				for (String dataSource : deleted) {
					monitor.validateDataSource(tenantId, JSONUtility.fromObjectString(dataSource));
					persistency.putDataSource(dataSource);
					producer.send(kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantId), dataSource.getBytes());
				}
				throw e;
			}
			String str = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, ServletUtil.constructCollectionJson(deleted));
			return Response.ok(str).type(MediaType.APPLICATION_JSON).build();

		} catch (Exception e) {
			e.printStackTrace();
		} catch (Throwable e) {
			e.printStackTrace();
		}
		String ret = ServletUtil.constructReturnMsg(200, "");
		return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();

	}

}
